{
 "cells": [
  {
   "metadata": {},
   "cell_type": "markdown",
   "source": [
    "下面是一个详细的 RxJava 学习示例，包含常用操作符和功能的代码实现，附带详细注释和日志打印：\n",
    "\n",
    "```java\n",
    "import io.reactivex.rxjava3.core.*;\n",
    "import io.reactivex.rxjava3.disposables.Disposable;\n",
    "import io.reactivex.rxjava3.functions.*;\n",
    "import io.reactivex.rxjava3.schedulers.Schedulers;\n",
    "import java.util.*;\n",
    "import java.util.concurrent.TimeUnit;\n",
    "\n",
    "public class RxJavaExample {\n",
    "\n",
    "    public static void main(String[] args) throws InterruptedException {\n",
    "        // 1. 创建型操作符\n",
    "        createObservable();\n",
    "        Thread.sleep(1000);\n",
    "\n",
    "        // 2. 转换操作符\n",
    "        transformOperators();\n",
    "        Thread.sleep(1000);\n",
    "\n",
    "        // 3. 过滤操作符\n",
    "        filterOperators();\n",
    "        Thread.sleep(1000);\n",
    "\n",
    "        // 4. 组合操作符\n",
    "        combiningOperators();\n",
    "        Thread.sleep(1000);\n",
    "\n",
    "        // 5. 错误处理\n",
    "        errorHandling();\n",
    "        Thread.sleep(1000);\n",
    "\n",
    "        // 6. 线程调度\n",
    "        threadingExample();\n",
    "        Thread.sleep(1000);\n",
    "\n",
    "        // 7. 背压处理\n",
    "        backpressureHandling();\n",
    "        Thread.sleep(5000); // 确保所有异步任务完成\n",
    "    }\n",
    "\n",
    "    // 1. 创建型操作符示例\n",
    "    private static void createObservable() {\n",
    "        System.out.println(\"\\n===== 创建型操作符 =====\");\n",
    "\n",
    "        // just: 直接发射预定义的数据项\n",
    "        Observable.just(\"Apple\", \"Banana\", \"Cherry\")\n",
    "                .subscribe(item -> System.out.println(\"just: \" + item));\n",
    "\n",
    "        // fromIterable: 从集合发射数据\n",
    "        List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);\n",
    "        Observable.fromIterable(numbers)\n",
    "                .subscribe(num -> System.out.println(\"fromIterable: \" + num));\n",
    "\n",
    "        // interval: 定时发射递增数字\n",
    "        Disposable intervalDisposable = Observable.interval(500, TimeUnit.MILLISECONDS)\n",
    "                .take(3) // 只取3个\n",
    "                .subscribe(\n",
    "                        tick -> System.out.println(\"interval: \" + tick),\n",
    "                        Throwable::printStackTrace,\n",
    "                        () -> System.out.println(\"interval 完成!\")\n",
    "                );\n",
    "    }\n",
    "\n",
    "    // 2. 转换操作符示例\n",
    "    private static void transformOperators() {\n",
    "        System.out.println(\"\\n===== 转换操作符 =====\");\n",
    "\n",
    "        // map: 类型转换\n",
    "        Observable.just(\"Hello\", \"World\")\n",
    "                .map(String::toUpperCase)\n",
    "                .subscribe(str -> System.out.println(\"map: \" + str));\n",
    "\n",
    "        // flatMap: 扁平化转换\n",
    "        Observable.just(\"A,B,C\", \"D,E,F\")\n",
    "                .flatMap(s -> Observable.fromArray(s.split(\",\")))\n",
    "                .subscribe(letter -> System.out.println(\"flatMap: \" + letter));\n",
    "\n",
    "        // concatMap: 保持顺序的扁平化\n",
    "        Observable.just(100, 200, 300)\n",
    "                .concatMap(delay -> Observable.just(delay)\n",
    "                        .delay(delay, TimeUnit.MILLISECONDS))\n",
    "                .subscribe(item -> System.out.println(\"concatMap: \" + item));\n",
    "\n",
    "        // buffer: 缓冲数据\n",
    "        Observable.range(1, 10)\n",
    "                .buffer(3)\n",
    "                .subscribe(list -> System.out.println(\"buffer: \" + list));\n",
    "    }\n",
    "\n",
    "    // 3. 过滤操作符示例\n",
    "    private static void filterOperators() {\n",
    "        System.out.println(\"\\n===== 过滤操作符 =====\");\n",
    "\n",
    "        // filter: 条件过滤\n",
    "        Observable.just(1, 2, 3, 4, 5, 6)\n",
    "                .filter(num -> num % 2 == 0)\n",
    "                .subscribe(even -> System.out.println(\"filter: \" + even));\n",
    "\n",
    "        // take: 取前N项\n",
    "        Observable.interval(200, TimeUnit.MILLISECONDS)\n",
    "                .take(3)\n",
    "                .subscribe(item -> System.out.println(\"take: \" + item));\n",
    "\n",
    "        // skip: 跳过前N项\n",
    "        Observable.range(1, 5)\n",
    "                .skip(2)\n",
    "                .subscribe(num -> System.out.println(\"skip: \" + num));\n",
    "\n",
    "        // distinct: 去重\n",
    "        Observable.just(1, 2, 2, 3, 3, 3)\n",
    "                .distinct()\n",
    "                .subscribe(unique -> System.out.println(\"distinct: \" + unique));\n",
    "    }\n",
    "\n",
    "    // 4. 组合操作符示例\n",
    "    private static void combiningOperators() {\n",
    "        System.out.println(\"\\n===== 组合操作符 =====\");\n",
    "\n",
    "        // zip: 多流合并\n",
    "        Observable<Integer> numbers = Observable.just(1, 2, 3);\n",
    "        Observable<String> letters = Observable.just(\"A\", \"B\", \"C\");\n",
    "\n",
    "        Observable.zip(numbers, letters,\n",
    "                (num, letter) -> num + letter)\n",
    "                .subscribe(comb -> System.out.println(\"zip: \" + comb));\n",
    "\n",
    "        // merge: 合并多个流（可能交错）\n",
    "        Observable<Long> fast = Observable.interval(100, TimeUnit.MILLISECONDS).take(3);\n",
    "        Observable<Long> slow = Observable.interval(300, TimeUnit.MILLISECONDS).take(2);\n",
    "\n",
    "        Observable.merge(fast, slow)\n",
    "                .subscribe(item -> System.out.println(\"merge: \" + item));\n",
    "\n",
    "        // concat: 顺序连接流\n",
    "        Observable.concat(\n",
    "                Observable.just(\"First\"),\n",
    "                Observable.just(\"Second\"),\n",
    "                Observable.just(\"Third\")\n",
    "        ).subscribe(item -> System.out.println(\"concat: \" + item));\n",
    "    }\n",
    "\n",
    "    // 5. 错误处理示例\n",
    "    private static void errorHandling() {\n",
    "        System.out.println(\"\\n===== 错误处理 =====\");\n",
    "\n",
    "        // onErrorReturn: 遇到错误返回默认值\n",
    "        Observable.error(new RuntimeException(\"模拟错误\"))\n",
    "                .onErrorReturn(throwable -> {\n",
    "                    System.out.println(\"捕获错误: \" + throwable.getMessage());\n",
    "                    return -1;\n",
    "                })\n",
    "                .subscribe(\n",
    "                        item -> System.out.println(\"onErrorReturn: \" + item),\n",
    "                        err -> System.err.println(\"这行不应该执行\")\n",
    "                );\n",
    "\n",
    "        // retry: 重试机制\n",
    "        AtomicInteger counter = new AtomicInteger();\n",
    "        Observable.create(emitter -> {\n",
    "                    if (counter.incrementAndGet() < 3) {\n",
    "                        emitter.onError(new RuntimeException(\"暂时错误\"));\n",
    "                    } else {\n",
    "                        emitter.onNext(\"成功数据\");\n",
    "                        emitter.onComplete();\n",
    "                    }\n",
    "                })\n",
    "                .retry(2) // 重试2次\n",
    "                .subscribe(\n",
    "                        data -> System.out.println(\"retry成功: \" + data),\n",
    "                        err -> System.out.println(\"最终失败: \" + err.getMessage())\n",
    "                );\n",
    "    }\n",
    "\n",
    "    // 6. 线程调度示例\n",
    "    private static void threadingExample() {\n",
    "        System.out.println(\"\\n===== 线程调度 =====\");\n",
    "\n",
    "        Observable.create(emitter -> {\n",
    "                    System.out.println(\"发射线程: \" + Thread.currentThread().getName());\n",
    "                    emitter.onNext(1);\n",
    "                    emitter.onComplete();\n",
    "                })\n",
    "                .subscribeOn(Schedulers.io()) // 指定发射线程\n",
    "                .observeOn(Schedulers.computation()) // 指定观察线程\n",
    "                .doOnNext(i -> System.out.println(\"处理线程: \" + Thread.currentThread().getName()))\n",
    "                .subscribe();\n",
    "    }\n",
    "\n",
    "    // 7. 背压处理示例\n",
    "    private static void backpressureHandling() {\n",
    "        System.out.println(\"\\n===== 背压处理 =====\");\n",
    "\n",
    "        Flowable.interval(10, TimeUnit.MILLISECONDS)\n",
    "                .onBackpressureBuffer(10, // 缓冲区大小\n",
    "                        () -> System.out.println(\"缓冲区溢出!\"),\n",
    "                        BackpressureOverflowStrategy.DROP_LATEST)\n",
    "                .observeOn(Schedulers.single(), false, 5) // 设置接收缓冲区大小\n",
    "                .subscribe(\n",
    "                        item -> {\n",
    "                            Thread.sleep(100); // 模拟慢消费者\n",
    "                            System.out.println(\"处理: \" + item);\n",
    "                        },\n",
    "                        Throwable::printStackTrace\n",
    "                );\n",
    "    }\n",
    "}\n",
    "```\n",
    "\n",
    "### 关键概念和操作符说明：\n",
    "\n",
    "1. **创建操作符**\n",
    "   - `just()`: 直接发射预定义数据\n",
    "   - `fromIterable()`: 从集合发射数据\n",
    "   - `interval()`: 定时发射递增数字\n",
    "\n",
    "2. **转换操作符**\n",
    "   - `map()`: 数据项一对一转换\n",
    "   - `flatMap()`: 一对多转换（顺序不保证）\n",
    "   - `concatMap()`: 保证顺序的一对多转换\n",
    "   - `buffer()`: 缓冲多个数据项\n",
    "\n",
    "3. **过滤操作符**\n",
    "   - `filter()`: 条件过滤\n",
    "   - `take()`: 取前N项\n",
    "   - `skip()`: 跳过前N项\n",
    "   - `distinct()`: 去重\n",
    "\n",
    "4. **组合操作符**\n",
    "   - `zip()`: 多流严格配对合并\n",
    "   - `merge()`: 多流合并（可能交错）\n",
    "   - `concat()`: 顺序连接多个流\n",
    "\n",
    "5. **错误处理**\n",
    "   - `onErrorReturn()`: 出错时返回默认值\n",
    "   - `retry()`: 遇到错误重试\n",
    "\n",
    "6. **线程控制**\n",
    "   - `subscribeOn()`: 指定发射线程\n",
    "   - `observeOn()`: 指定下游处理线程\n",
    "\n",
    "7. **背压处理**\n",
    "   - `onBackpressureBuffer()`: 设置缓冲区策略\n",
    "   - `Flowable`: 专为背压设计的可观察对象\n",
    "\n",
    "### 日志输出示例：\n",
    "```\n",
    "===== 创建型操作符 =====\n",
    "just: Apple\n",
    "just: Banana\n",
    "just: Cherry\n",
    "fromIterable: 1\n",
    "fromIterable: 2\n",
    "...\n",
    "interval: 0\n",
    "interval: 1\n",
    "interval: 2\n",
    "interval 完成!\n",
    "\n",
    "===== 转换操作符 =====\n",
    "map: HELLO\n",
    "map: WORLD\n",
    "flatMap: A\n",
    "flatMap: B\n",
    "...\n",
    "concatMap: 100\n",
    "concatMap: 200\n",
    "concatMap: 300\n",
    "buffer: [1, 2, 3]\n",
    "...\n",
    "```\n",
    "\n",
    "### 使用建议：\n",
    "1. **依赖配置** (Maven):\n",
    "```xml\n",
    "<dependency>\n",
    "    <groupId>io.reactivex.rxjava3</groupId>\n",
    "    <artifactId>rxjava</artifactId>\n",
    "    <version>3.1.8</version>\n",
    "</dependency>\n",
    "```\n",
    "\n",
    "2. **运行说明**：\n",
    "- 示例中的 `Thread.sleep()` 是为了防止主线程提前退出\n",
    "- 实际开发中应使用 `Disposable` 管理资源\n",
    "- 背压场景优先使用 `Flowable`\n",
    "\n",
    "3. **调试技巧**：\n",
    "- 添加 `doOnNext()`, `doOnError()` 等生命周期钩子\n",
    "- 使用 `compose()` 封装通用操作链\n",
    "- 通过 `observeOn()` 控制不同阶段的线程\n",
    "\n",
    "这个示例覆盖了 RxJava 的核心功能，通过日志可以清晰看到数据流的变化过程。建议从简单操作符开始逐步实验，理解响应式编程的\"数据流\"思想。"
   ]
  },
  {
   "metadata": {},
   "cell_type": "code",
   "outputs": [],
   "execution_count": null,
   "source": ""
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Kotlin",
   "language": "kotlin",
   "name": "kotlin"
  },
  "language_info": {
   "name": "kotlin",
   "version": "1.9.23",
   "mimetype": "text/x-kotlin",
   "file_extension": ".kt",
   "pygments_lexer": "kotlin",
   "codemirror_mode": "text/x-kotlin",
   "nbconvert_exporter": ""
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}
